package david.java.flink.example;

import david.java.flink.datasources.Tuple3Source;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.Tuple3;

/**
 * @Description:  args   --test hahaha --name 范老师
 * @Author: Zhaodawei
 * @Date: Create in 下午1:21 2020/12/18
 */
public class WordCount {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setString("conf", "123456");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

        //参数全局化
        ParameterTool params = ParameterTool.fromArgs(args);
        env.getConfig().setGlobalJobParameters(params);
        // Configuration configuration = new Configuration();


        DataStreamSource<Tuple3<String, Long, Long>> data = env.addSource(new Tuple3Source());
        SingleOutputStreamOperator<Object> map = data.map(x -> {
            System.out.println("print\t" + x._1() + "\t" + x._2() + "\t" + x._3());
            return x;
        });

        map.map(new RichMapFunction<Object, Object>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ParameterTool globalParams = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

                String test = parameters.getString("test", "default");
                System.out.println(test);
                System.out.println(parameters.getString("name", "nonono"));
                System.out.println(globalParams.get("test", "default"));
                System.out.println(globalParams.get("name", "nonono"));
                System.out.println(globalParams.get("name", "nonono"));
            }

            @Override
            public Object map(Object value) throws Exception {
                return value;
            }
        }).print();

        // map.print();

        env.execute("word count");
    }
}
